package org.jeecg.modules.rental.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.shiro.SecurityUtils;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.exception.JeecgBootException;
import org.jeecg.common.rental.vo.LoginRentalAccount;
import org.jeecg.common.util.DateUtils;
import org.jeecg.common.util.JacksonUtil;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.jt808.Jt808Control.Jt808CentralControlConfig;
import org.jeecg.jt808.modbus.ModbusParams;
import org.jeecg.jt808.processor.JT808MessageProcessor;
import org.jeecg.jt808.processor.MessageType;
import org.jeecg.modules.iot.api.IControlAPI;
import org.jeecg.modules.iot.model.LocationJson;
import org.jeecg.modules.iot.model.control.CentralControlConfig;
import org.jeecg.modules.iot.model.control.SpeedOtherConfig;
import org.jeecg.modules.iot.util.DouglasPeucker;
import org.jeecg.modules.rental.entity.RentalAccount;
import org.jeecg.modules.rental.entity.RentalAccountPackage;
import org.jeecg.modules.rental.entity.RentalEBike;
import org.jeecg.modules.rental.mapper.RentalEBikeMapper;
import org.jeecg.modules.rental.req.EBikeTrackReq;
import org.jeecg.modules.rental.service.ICtrlEBikeService;
import org.jeecg.modules.rental.service.IRentalAccountPackageService;
import org.jeecg.modules.rental.service.IRentalAccountService;
import org.jeecg.modules.rental.service.IRentalEBikeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.List;

@Slf4j
@Service
@RequiredArgsConstructor
public class CtrlEBikeService implements ICtrlEBikeService {

    private final IRentalAccountPackageService rentalAccountPackageService;
    private final IRentalAccountService rentalAccountService;
    private final RentalEBikeMapper rentalCarMapper;
    private final IControlAPI controlApi;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final RedisUtil redisUtil;
    private static final String NATIONAL_STANDARD_SPEED_LIMIT_KEY = "vehicle:%s:nationalStandardSpeedLimit"; //  Redis key 格式

    //============================== 基于租车套餐的操作 ==============================

    @Override
    public Boolean powerOnByPkg() {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.OPEN_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean powerOffByPkg() {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.CLOSE_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean lockedByPkg() {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.DEFENSE_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean unlockByPkg() {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.DISARMING_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean searchByPkg() {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SEARCH_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean unlockSeatByPkg() {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SEAT_LOCK);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean setSportModeByPkg(Boolean enable) {
        //TODO 暂时按照单套餐处理
        RentalAccountPackage accountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(accountPackage.getCarId());
        Boolean result = false;
        try {
            String sportModeRedisKey = String.format(NATIONAL_STANDARD_SPEED_LIMIT_KEY, car.getCarCode());
            redisUtil.set(sportModeRedisKey, enable);
            Jt808CentralControlConfig config = new Jt808CentralControlConfig();
            config.setSpeedLimitEnabled(enable); // 设置限速开关
            result=   sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SET_SPEED_LIMIT, config, null);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public String trackListBypkg(EBikeTrackReq trackReq) {
        if (ObjectUtil.isNotEmpty(trackReq.getStartTime()) && ObjectUtil.isNotEmpty(trackReq.getEndTime())) {
            if (trackReq.getStartTime().isAfter(trackReq.getEndTime())) {
                throw new JeecgBootException("开始时间不能晚与结束时间");
            }
        }

        RentalAccountPackage rentalAccountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        if (ObjectUtil.isEmpty(rentalAccountPackage)) {
            throw new JeecgBootException("没有找到对应的租车套餐");
        }
        if (ObjectUtil.isEmpty(rentalAccountPackage.getCarId())) {
            throw new JeecgBootException("租车套餐没有绑定电单车");
        }
        RentalEBike rentalCar = rentalCarMapper.selectById(rentalAccountPackage.getCarId());
        if (ObjectUtil.isEmpty(rentalCar)) {
            throw new JeecgBootException("未找到对应电单车信息");
        }

        return this.trackList(trackReq, rentalCar.getCarCode());
    }

    @Override
    public Boolean startPushByPkg() {
        RentalAccountPackage rentalAccountPackage = rentalAccountPackageService.getCurrentUserAccountPackage();
        RentalEBike car = rentalCarMapper.selectById(rentalAccountPackage.getCarId());
        return controlApi.pushEBikeInfo(car.getCarCode());
    }

    //============================== 基于租车账户(绑定车辆)的操作 ==============================

    @Override
    public Boolean powerOnByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.OPEN_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean powerOffByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.CLOSE_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean lockedByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.DEFENSE_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean unlockByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.DISARMING_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean searchByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SEARCH_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean unlockSeatByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SEAT_LOCK);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean setSportModeByAccount(Boolean enable) {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        Boolean result = false;
        try {
            /*result = controlApi.enableSportMode(car.getCarCode(), car.getPhone(), enable);*/
            String sportModeRedisKey = String.format(NATIONAL_STANDARD_SPEED_LIMIT_KEY, car.getCarCode());
            redisUtil.set(sportModeRedisKey, enable);
            Jt808CentralControlConfig config = new Jt808CentralControlConfig();
            config.setSpeedLimitEnabled(enable); // 设置限速开关
            result=   sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SET_SPEED_LIMIT, config, null);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public String trackListByAccount(EBikeTrackReq trackReq) {
        if (ObjectUtil.isNotEmpty(trackReq.getStartTime()) && ObjectUtil.isNotEmpty(trackReq.getEndTime())) {
            if (trackReq.getStartTime().isAfter(trackReq.getEndTime())) {
                throw new JeecgBootException("开始时间不能晚与结束时间");
            }
        }
        RentalAccount rentalAccount = rentalAccountService.getCurrentUserAccount();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        if (ObjectUtil.isEmpty(car)) {
            throw new JeecgBootException("账户没有绑定电单车");
        }
        return this.trackList(trackReq, car.getCarCode());
    }


    @Override
    public Boolean startPushByAccount() {
        LoginRentalAccount rentalAccount = (LoginRentalAccount) SecurityUtils.getSubject().getPrincipal();
        RentalEBike car = rentalCarMapper.selectShareEBike(rentalAccount.getId());
        try{
            return controlApi.pushEBikeInfo(car.getCarCode());
        }catch(Exception e){}
        return false;
    }

    //============================== 直接操作车辆 ==============================

    @Override
    public Boolean powerOn(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.OPEN_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean powerOff(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.CLOSE_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean locked(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.DEFENSE_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean unlock(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.DISARMING_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean search(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SEARCH_CAR);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean unlockSeat(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SEAT_LOCK);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public Boolean lockOpen(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result=false;
        try {
//            result = controlApi.handleLockOpen(car.getCarCode(), car.getPhone());
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.HANDLE_LOCK_OPEN);
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Boolean lockClose(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result=false;
        try {
//            result = controlApi.handleLockClose(car.getCarCode(), car.getPhone());
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.HANDLE_LOCK_CLOSE);
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Boolean setSportMode(String carCode, Boolean enable) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result = false;
        try {
            String sportModeRedisKey = String.format(NATIONAL_STANDARD_SPEED_LIMIT_KEY, car.getCarCode());
            redisUtil.set(sportModeRedisKey, enable);
            Jt808CentralControlConfig config = new Jt808CentralControlConfig();
            config.setSpeedLimitEnabled(enable); // 设置限速开关
            result = sendCommandViaKafka(car.getCarCode(), car.getPhone(), MessageType.SET_SPEED_LIMIT, config, null);
        } catch (Exception e) {
            throw new JeecgBootException(e);
        }
        return result;
    }

    @Override
    public String trackList(EBikeTrackReq trackReq, String carCode) {
        String pathJson = "";
        //默认时间范围
        LocalDateTime startTime = LocalDate.now().atStartOfDay();
        LocalDateTime endTime = LocalDate.now().atTime(LocalTime.MAX);

        // 如果 startTime 为空，设置 startTime 为当天的开始时间
        if (ObjectUtil.isNotEmpty(trackReq.getStartTime())) {
            startTime = trackReq.getStartTime();
        }

        // 如果 endTime 为空，设置 endTime 为当前时间
        if (ObjectUtil.isNotEmpty(trackReq.getEndTime())) {
            endTime = trackReq.getEndTime();
        }

        //TODO 查询轨迹
        Timestamp startTimestamp = DateUtils.convertToTimestamp(startTime);
        Timestamp endTimestamp = DateUtils.convertToTimestamp(endTime);

        ArrayNode arrayNode = JacksonUtil.newJSONArray();
        try{
            List<LocationJson> carTrackList = controlApi.queryLocationsByTimeRange(carCode, startTimestamp, endTimestamp);
            List<LocationJson> simplifiedTrackList = DouglasPeucker.simplify(carTrackList, 0.0001);
            simplifiedTrackList.stream().forEach(item->{
                ObjectNode objectNode = JacksonUtil.newJSONObject();
                objectNode.put("lat", item.getLat());
                objectNode.put("lng", item.getLng());
                objectNode.put("speed", item.getSpeed());
                objectNode.put("mil", item.getMileage());
                arrayNode.add(objectNode);
            });
        }catch(Exception ex){
            throw new JeecgBootException("common.query.not.found");
        }
        return JacksonUtil.toJSONString(arrayNode);
    }

    @Override
    public Boolean getSportModeStatus(String carCode, String phone) {
        return controlApi.getSportModeStatus(carCode, phone);
    }

    @Override
    public Boolean pairRemoteControl(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result=false;
        try {
            result = controlApi.pairRemoteControl(car.getCarCode(), car.getPhone());
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Boolean clearRemoteControl(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result=false;
        try {
            result = controlApi.clearRemoteControl(car.getCarCode(), car.getPhone());
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Boolean setVolume(String carCode, Integer volume) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result=false;
        try {
            result = controlApi.setVolume(car.getCarCode(),car.getPhone(),volume);
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Boolean setPlayAudio(String carCode, Byte audioIndex) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Boolean result=false;
        try {
            result = controlApi.setPlayAudio(car.getCarCode(),car.getPhone(),audioIndex);
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Integer setPower(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Integer result = 0;
        try {
            result = controlApi.setPower(car.getCarCode(),car.getPhone(),true,1,20);
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Integer queryPower(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Integer result = 0;
        try {
            result = controlApi.queryPower(car.getCarCode(),car.getPhone());
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Result setSpeed(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Result result;
        try {
            result = controlApi.setSpeed(car.getCarCode(),car.getPhone(), new SpeedOtherConfig());
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public String queryControl(String carCode) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        String result = "";
        try {
            result = controlApi.queryControl(car.getCarCode(),car.getPhone());
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    @Override
    public Result setControl(String carCode, CentralControlConfig config) {
        RentalEBike car = rentalCarMapper.selectOne(new LambdaQueryWrapper<RentalEBike>().eq(RentalEBike::getCarCode, carCode));
        Result result;
        try {
            result = controlApi.setCentralControlConfig(car.getCarCode(),car.getPhone(), config);
        }  catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    private boolean sendCommandViaKafka(String carCode, String phone, MessageType messageType) {
        try {
            if (carCode == null || phone == null) {  // 校验参数
                log.warn("carCode 或 phone 为空，无法发送指令");
                return false;
            }
            JT808MessageProcessor jt808MessageProcessor = new JT808MessageProcessor();
            String controlStr = jt808MessageProcessor.process(carCode, phone, messageType);
            kafkaTemplate.send("device_send", carCode, controlStr);
            return true;
        } catch (Exception e) {
            log.error("发送 Kafka 指令失败", e);
            return false;
        }
    }

    private boolean sendCommandViaKafka(String carCode, String phone, MessageType messageType, Jt808CentralControlConfig config, ModbusParams modbusParams) {
        try {
            if (carCode == null || phone == null) {
                log.warn("carCode 或 phone 为空，无法发送指令");
                return false;
            }
            JT808MessageProcessor jt808MessageProcessor = new JT808MessageProcessor();
            String controlStr = jt808MessageProcessor.process(carCode,phone, messageType,   config) ;
            kafkaTemplate.send("device_send", carCode, controlStr);
            return true;
        } catch (Exception e) {
            log.error("发送 Kafka 指令失败", e);
            return false;
        }
    }
}
